package com.example.vpnservicedemo.receive;


import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;


public class TcpSelector {
    private static Selector selector;

    private static ConcurrentLinkedQueue<RegisterData> registerQueue = new ConcurrentLinkedQueue<>();

    static {

        try {
            selector = Selector.open();
        } catch (IOException e) {
            System.out.println("TcpSelector 启动失败:" + e.getMessage());
        }

        Thread thread = new Thread(() -> {
            try {
                while (!Thread.interrupted()) {
                    // 等待事件
                    selector.select();

                    // 获取事件的迭代器
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();

                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();

                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        TcpPipe tcpPiping = (TcpPipe) key.attachment();

                        if (key.isConnectable()) {

                            try {
                                // 完成连接[这里会出现超时异常]
                                socketChannel.finishConnect();

                                //切换事件
                                key.interestOps(SelectionKey.OP_READ);

                                tcpPiping.replySyn();

                            } catch (Exception ex) {
                                ex.printStackTrace();
                                tcpPiping.close();
                            }


                        } else if (key.isReadable()) {

                            try {

                                ByteBuffer buffer = ByteBuffer.allocate(1024);
                                int bytesRead = socketChannel.read(buffer);

                                if (bytesRead == -1) {
                                    socketChannel.close();
                                    //推送一个结束
                                    tcpPiping.close();

                                } else if (bytesRead > 0) {
                                    buffer.flip();
                                    byte[] array = new byte[buffer.limit()];
                                    buffer.get(array);

                                    tcpPiping.replyPsh(array);

                                }

                            } catch (Exception e) {
                                tcpPiping.close();
                                e.printStackTrace();
                            }

                        }

                        // 删除处理过的事件
                        iterator.remove();
                    }
                    RegisterData registerData = null;
                    while ((registerData = registerQueue.poll()) != null) {
                        registerData.socketChannel.register(selector, SelectionKey.OP_CONNECT, registerData.tcpPipe);
                    }

                }

            } catch (Exception e) {
                System.out.println("TcpSelector 线程异常结束:" + e.getMessage());
                e.printStackTrace();
            }

        });
        thread.setName("tcpSelector");
        thread.start();

    }

    public static void registerConnect(SocketChannel socketChannel, TcpPipe tcpPipe) {

        RegisterData registerData = new RegisterData();
        registerData.socketChannel = socketChannel;
        registerData.tcpPipe = tcpPipe;

        registerQueue.offer(registerData);
        selector.wakeup();
    }

    static class RegisterData {
        SocketChannel socketChannel;
        TcpPipe tcpPipe;
    }


}
